package test.wordcount;

import java.io.IOException;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionBuilder;
import org.apache.commons.cli.OptionGroup;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import cache.io.CacheInputFormat;

public class WordCount {
	private static final Log LOG = LogFactory.getLog(WordCount.class);
	
	public static final String OPT_INPUT_TABLE    = "t"; // Use a database
	public static final String OPT_INPUT_FILE     = "f"; // Use an input file
	
	public static final String OPT_OUTPUT         = "o";
	public static final String OPT_NUM_OF_MAPPERS = "m";
	public static final String OPT_USE_CACHE      = "s";
	
	static Options options;
	
	
	@SuppressWarnings("static-access")
	private static void setupOptions() {
		
		options = new Options();
		
		/*
		 * General options not related to intput. 
		 */
		Option outputFilename = OptionBuilder.withArgName("filename")
			.withDescription("output file")
			.hasArg()
			.isRequired()
			.create(OPT_OUTPUT);

		Option numOfMappers = OptionBuilder.withArgName("number")
			.withDescription("number of mappers")
			.hasArg()
			.isRequired()
			.create(OPT_NUM_OF_MAPPERS);
		
		Option useCache = OptionBuilder
			.withDescription("use cache")
			.create(OPT_USE_CACHE);
		
		/*
		 * Input options
		 */
		Option tableName = OptionBuilder
			.withArgName("table")
			.withDescription("input table name")
			.hasArg()
			.create(OPT_INPUT_TABLE);
		
		Option inputFileName = OptionBuilder
			.withArgName("filename")
			.withDescription("input file name")
			.hasArg()
			.create(OPT_INPUT_FILE);
		
		OptionGroup inputOptions = new OptionGroup();
		inputOptions.addOption(tableName);
		inputOptions.addOption(inputFileName);
		inputOptions.setRequired(true);
		

		options.addOption(outputFilename);
		options.addOption(numOfMappers);
		options.addOption(useCache);

		options.addOptionGroup(inputOptions);
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		setupOptions();
		CommandLine line = null;
		
	    try {
	    	CommandLineParser parser = new GnuParser();
	        line = parser.parse( options, args );
	    }
	    catch( ParseException exp ) {
		    HelpFormatter formatter = new HelpFormatter();
		    formatter.printHelp( "WordCount", options );
		    System.exit(-1);
	    }
	    
		Job job = new Job();
		job.setJarByClass(WordCount.class);
		job.setJobName("Word-Count " + ArrayUtils.toString(args));

		//FileSystem fs = FileSystem.getLocal(job.getConfiguration());
		FileSystem fs = FileSystem.get(job.getConfiguration());
		Path outputPath = new Path(line.getOptionValue(OPT_OUTPUT));
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}
		
		setup(line, job);
        
		FileOutputFormat.setOutputPath(job, outputPath);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		job.setReducerClass(WordCountReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		long startTime = System.currentTimeMillis();
		boolean success = job.waitForCompletion(true); 		
		long endTime = System.currentTimeMillis();
		LOG.info("Running time (ms): " + (endTime - startTime));
		System.exit( success ? 0 : 1);
	}

	/**
	 * @param line
	 * @param job
	 * @throws IOException 
	 */
	private static void setup(CommandLine line, Job job) throws IOException {
		job.getConfiguration().setBoolean("mapred.mapper.new-api", true);
		
		int numOfMappers = 2;
		try {
			numOfMappers = Integer.parseInt(line.getOptionValue(OPT_NUM_OF_MAPPERS));
		} catch (NumberFormatException e) {
			System.err.println("Number of mappers is not ok.");
			System.exit(-1);
		}
		job.getConfiguration().setInt("mapred.map.tasks", numOfMappers);

		if (line.hasOption(OPT_INPUT_TABLE)) {
			String username = "dbs131_user25";
			// String username = "root";
			String password = "dbsdbs";
			// String password = "";
			String connectionString = "jdbc:mysql://mysqlsrv1/dbs131_user25";
			//String connectionString = "jdbc:mysql://127.0.0.1/hadoop_cache";
			
			String query = "select SQL_NO_CACHE * from " + line.getOptionValue(OPT_INPUT_TABLE);
			String inputFormatId = connectionString + "|" + query;

			DBConfiguration.configureDB (
					job.getConfiguration(), 
					"com.mysql.jdbc.Driver",
	                connectionString, username, password
	        );
			
			String[] fields = { "line" }; //Specifies the Fields to be fetched from DB
			
			/*
	        DBInputFormat.setInput(
	        		job, 
	        		TextRecord.class, 
	        		line.getOptionValue(OPT_INPUT_TABLE), // table name
	                null, // conditions
	                null, 
	                fields); // Specifies the DB table and fields
	        */
	        
	        DBInputFormat.setInput(
	        		job, 
	        		TextRecord.class,
	        		query, /* table name */
	        		"select count(*) from " + line.getOptionValue(OPT_INPUT_TABLE) /* table name */
	        		);
	        
			job.setMapperClass(WordCountMapperDb.class);
			
			if (line.hasOption(OPT_USE_CACHE)) {
				job.setInputFormatClass(CacheInputFormat.class);
				CacheInputFormat.setDelegateInputFormatData(job, DBInputFormat.class, inputFormatId);
			} else {
				job.setInputFormatClass(DBInputFormat.class);
			}			
			
		} else {
			/*
			 * Using a file
			 */
			String inputFilename = line.getOptionValue(OPT_INPUT_FILE);
			FileInputFormat.addInputPath(job, new Path(inputFilename));
			job.setMapperClass(WordCountMapper.class);
			
			if (line.hasOption(OPT_USE_CACHE)) {
				job.setInputFormatClass(CacheInputFormat.class);
				CacheInputFormat.setDelegateInputFormatData(job, TextInputFormat.class, inputFilename);
			} else {
				job.setInputFormatClass(TextInputFormat.class);
			}			
			
			
			
		}
		

	}
}
